package com.xydpeng.rpc.protocol.handler;

import com.xydpeng.rpc.common.model.FastRpcRequest;
import com.xydpeng.rpc.common.model.FastRpcResponse;
import com.xydpeng.rpc.protocol.FastRpcProtocol;
import com.xydpeng.rpc.protocol.HandlerInvoker;
import com.xydpeng.rpc.protocol.HandlerMapping;
import com.xydpeng.rpc.protocol.HandlerMeta;
import com.xydpeng.rpc.protocol.MsgHeader;
import com.xydpeng.rpc.protocol.MsgStatus;
import com.xydpeng.rpc.protocol.MsgType;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;

/**
 * @author xydpeng
 * @version v1.0
 * @date: 2023/4/13
 * @description:
 */
@Slf4j
public class RpcRequestHandler extends SimpleChannelInboundHandler<FastRpcProtocol<FastRpcRequest>> {
    private final ExecutorService businessThreadpool;

    public RpcRequestHandler(ExecutorService businessThreadpool) {
        this.businessThreadpool = businessThreadpool;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FastRpcProtocol<FastRpcRequest> msg) throws Exception {
        businessThreadpool.submit(() -> {
            handle(ctx, msg);
        });
    }


    private void handle(ChannelHandlerContext ctx, FastRpcProtocol msg) {
        MsgHeader header = msg.getHeader();
        FastRpcRequest request = (FastRpcRequest) msg.getBody();
        int handlerId = request.getHandlerId();
        FastRpcResponse<Object> response = new FastRpcResponse<>();
        byte status = MsgStatus.SUCCESS.getCode();
        Object reslut = null;
        try {
            HandlerMeta handlerMeta = HandlerMapping.getHandlerMeta(handlerId);
            if (handlerMeta == null) {
                status = MsgStatus.FAIL.getCode();
            }
            HandlerInvoker handlerInvoker = handlerMeta.getInvoker();
            reslut = handlerInvoker.invoke(handlerId, request.getParams());
        } catch (Exception e) {
            status = MsgStatus.FAIL.getCode();
            response.setMsg(e.toString());
        }
        header.setStatus(status);
        header.setMsgType(MsgType.RESPONSE.getType());
        response.setData(reslut);
        final FastRpcProtocol resProtocol = msg;
        resProtocol.setHeader(header);
        resProtocol.setBody(response);
        ctx.writeAndFlush(resProtocol);
    }

}
